package com.pig.consume.invoker;

import com.pig.common.*;
import com.pig.common._enum.ProtocolEnum;
import com.pig.consume.client.ResultManager;
import com.pig.consume.client.TextClient;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Type;
import java.util.UUID;
import java.util.concurrent.*;


/**
 * @author lizhiwei
 */
@Slf4j
public class ConsumeProxy implements InvocationHandler {
    /**
     * 此线程池用来轮询请求结果
     */
    private static ExecutorService executorNettyThread = Executors.newFixedThreadPool(20);
    /**
     * netty客户端
     */
    private TextClient textClient;

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Exception {
        try {
            if (method == null) {
                throw new Exception("not found method");
            }
            log.info("proxy={},method={},args={}", new Object[]{proxy.getClass(), method, args});
            String className = method.getDeclaringClass().getName();
            ProtocolEnum protocolEnum = RpcContext.get(className);
            //默认为http协议
            if (protocolEnum == null) {
                protocolEnum = ProtocolEnum.HTTP;
            }
            switch (protocolEnum) {
                case HTTP:
                    Object httpObj = http(method, args);
                    return httpObj;
                case NETTY:
                    Object nettyObj = netty(method, args);
                    return nettyObj;
                default:
                    throw new Exception("not found protocol");
            }
        } catch (IOException e) {
            log.error("call rpc error", e);
            e.printStackTrace();
        }
        return null;
    }

    private Object http(Method method, Object[] args) throws IOException {
        String methodHash = MethodUtils.methodEncode(method);
        String argsStr = MethodUtils.argEncode(args);
        String url = "http://localhost:8081/provider?" + RpcConst.REQUEST_INFO + "=" + methodHash + "&" + RpcConst.ARGS + "=" + argsStr;
        String requestResult = HttpUtils.post(url, "");
        log.info("http success,{}", requestResult);
        Type genericReturnType = method.getGenericReturnType();
        return JsonUtils.jsonStrToBean(requestResult, genericReturnType);
    }

    private Object netty(Method method, Object[] args) throws Exception {
        initTextClient();
        String methodHash = MethodUtils.methodEncode(method);
        String argsStr = MethodUtils.argEncode(args);
        //使用uuid作为消息传输的key
        String uuid = UUID.randomUUID().toString();
        String param = uuid + "&" + methodHash + "&" + argsStr;
        textClient.sendMsg(param);
        Future<String> submit = executorNettyThread.submit(() -> {
            while (true) {
                String result = ResultManager.get(uuid);
                if (result == null) {
                    Thread.sleep(10L);
                } else {
                    return result;
                }
            }
        });
        //这里会阻塞，直到返回结果
        String result = null;
        try {
            result = submit.get(1000L, TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            throw new Exception("未能得到请求结果");
        }
        Type genericReturnType = method.getGenericReturnType();
        log.info("netty success,{}", result);
        return JsonUtils.jsonStrToBean(result, genericReturnType);
    }

    private void initTextClient() {
        if (textClient == null) {
            textClient = new TextClient("localhost", 9001);
            try {
                textClient.run();
                Thread.sleep(1000L);
            } catch (Exception e) {
                log.error("start netty error!");
            }
        }
    }

}
